package cn.whuc.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Demo_CategoryTop10_4 {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    // 2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19

    /*
      思路4 使用累加器实现
      累加器 输入 cid和类型
            输出 Category样例类对象的map集合
     */
    val datas: RDD[String] = sc.textFile("input/user_visit_action.txt")

    val acc = new HotCategoryAccumulator

    sc.register(acc)

    datas.foreach(line => {
      val strings: Array[String] = line.split("_")
      if (strings(6) != "-1") {
        acc.add(strings(6), "click")
      } else if (strings(8) != "null") {
        val ids: Array[String] = strings(8).split(",")
        ids.map(id => {
          acc.add(id, "order")
        })
      } else if (strings(10) != "null") {
        val ids: Array[String] = strings(10).split(",")
        ids.map(id => {
          acc.add(id, "pay")
        })
      }
    })

    //    println(acc.value)

    // 对 category对象进行比较排序 样例类 默认就继承了 比较接口 好用
    val result: List[Category] = acc.value.map(_._2).toList.sortWith(
      (l, r) => {
        if (l.clickcount > r.clickcount) {
          true // 逆序 排列
        } else if (l.clickcount == r.clickcount) {
          if (l.ordercount > r.ordercount) {
            true
          } else if (l.ordercount == r.ordercount) {
            if (l.paycount > r.paycount) {
              true
            } else {
              false
            }
          } else {
            false
          }
        } else {
          false
        }
      }
    ).take(10)

    println(result)

    // 3 关闭上下文对象
    sc.stop()
  }

}

case class Category(cid: String, var clickcount: Int, var ordercount: Int, var paycount: Int)

// 累加器
class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, Category]] {
  private var map = mutable.Map[String, Category]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[(String, String), mutable.Map[String, Category]] = new HotCategoryAccumulator()

  override def reset(): Unit = map.clear()

  override def add(v: (String, String)): Unit = {
    val cid: String = v._1
    val actionType: String = v._2
    val category: Category = map.getOrElse(cid, Category(cid, 0, 0, 0))
    if (actionType == "click") {
      category.clickcount += 1
    } else if (actionType == "order") {
      category.ordercount += 1
    } else if (actionType == "pay") {
      category.paycount += 1
    }
    map.update(cid, category)
  }

  override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, Category]]): Unit = {
    val map1: mutable.Map[String, Category] = this.map
    val map2: mutable.Map[String, Category] = other.value
    map2.foreach {
      case (cid, category) => {
        val category1: Category = map1.getOrElse(cid, Category(cid, 0, 0, 0))
        category1.clickcount += category.clickcount
        category1.ordercount += category.ordercount
        category1.paycount += category.paycount
        map1.update(cid, category1)
      }
    }
  }

  override def value: mutable.Map[String, Category] = map
}

